package com.steakliu.sun.buffer.disuptor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.IgnoreExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.steakliu.sun.buffer.DisruptorDataBuffer;
import com.steakliu.sun.protocol.SunProtocol;

/**
 * 功能说明：
 * <p>
 * Original @Author: steakliu-刘牌, 2023-02-09  16:22
 * <p>
 * Copyright (C)2020-2022  steakliu All rights reserved.
 */
public class DataEventPublisher<T> {

    private final RingBuffer<DataEvent<T>> ringBuffer;

    DataEventPublisher() {
        DataEventFactory<T> factory = new DataEventFactory<>();
        int bufferSize = 2 * 2048;
        Disruptor<DataEvent<T>> disruptor = new Disruptor<>(factory, bufferSize, r -> {
            return new Thread(null,r, "message-publisher-thread");
        }, ProducerType.MULTI, new BlockingWaitStrategy());

        disruptor.handleEventsWith(new DisruptorDataBuffer<T>());
        disruptor.setDefaultExceptionHandler(new IgnoreExceptionHandler());
        ringBuffer = disruptor.getRingBuffer();
        disruptor.start();
    }

    public void publishEvent(SunProtocol<T> protocol) {
        long sequence = ringBuffer.next();
        try {
            DataEvent<T> event = ringBuffer.get(sequence);
            event.setProtocol(protocol);
        } finally {
            ringBuffer.publish(sequence);
        }
    }
}
